package rocketMQ;

import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.consumer.MessageQueueListener;
import org.apache.rocketmq.client.consumer.TopicMessageQueueChangeListener;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Set;

import static rocketMQ.MqConstant.MQ_ADDRESS;

/**
 * 一些消费者
 *
 * @author liuchaoqun
 * @date 2023/02/15
 */
public class SomeConsumer {

    public static void main(String[] args) throws MQClientException {
        DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("cg");
        consumer.setNamesrvAddr(MQ_ADDRESS);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET );
        consumer.subscribe("someTopic", "*");
        MessageListenerConcurrently messageListenerConcurrently = (list, consumeConcurrentlyContext) -> {
            for (MessageExt messageExt : list) {
                System.out.println(messageExt);
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        };

        consumer.start();
        while (consumer.isRunning()) {
            List<MessageExt> message = consumer.poll();
            for (MessageExt messageExt : message) {
                System.out.println(new String(messageExt.getBody(), StandardCharsets.UTF_8));

            }
        }

        System.out.println("start");


    }
}
